package com.project.KafkaRoute.executor;

import com.project.MessageRoute.executor.PushConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.function.Function;

import static java.lang.Boolean.FALSE;

public class KafkaPushConsumer implements PushConsumer {

    private KafkaConsumer<String, String> consumers;
    private List<String> topics = new ArrayList<>();
    private long poll_time;
    private Thread run;
    private boolean exec = true;

    public KafkaPushConsumer(
            String name,
            String servers,
            String reset,
            long poll,
            Function<ConsumerRecords<String,String>, Boolean> function) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, name);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, reset);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, FALSE);
        this.poll_time = poll;

        consumers = new KafkaConsumer<>(props);

        run = new Thread(()->{
            while(exec) {
                ConsumerRecords<String,String> record = consumers.poll(Duration.ofMillis(poll_time));
                if (function.apply(record)) {
                    consumers.commitAsync();
                }
            }
        });
    }

    @Override
    public void subscribe(String topic, String subExpression) {
        topics.add(topic);
    }

    @Override
    public void start() {
        exec = true;
        consumers.subscribe(topics);
        run.start();
    }

    @Override
    public void close() {
        exec = false;
    }
}
